package com.xj.testcontainers;

import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.file.FileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.BindMode;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
import org.testcontainers.lifecycle.Startable;
import org.testcontainers.shaded.org.apache.commons.io.FilenameUtils;

import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

import static java.time.temporal.ChronoUnit.SECONDS;

/**
 * 自定义的docker测试容器类，需要实现接口
 */
public class CitusCluster implements Startable {

    private GenericContainer postgres;
    private List<GenericContainer> workerContainers = new ArrayList<>();

    public CitusCluster() {
        Network network = Network.newNetwork();
        String dockerImageName = "citusdata/citus:8.1.1";
        this.postgres = new GenericContainer(dockerImageName)
                .withExposedPorts(5432)
                .waitingFor(new LogMessageWaitStrategy()
                        .withRegEx(".*database system is ready to accept connections.*\\s")
                        .withTimes(2)
                        .withStartupTimeout(Duration.of(60, SECONDS)))
                .withCommand("postgres", "-c", "fsync=off")     //关闭文件同步
                .withNetwork(network);

        //创建集群
        for (int i = 0; i < 2; i++) {
            GenericContainer citus = new GenericContainer(dockerImageName)
                    .withExposedPorts(5432)
                    .waitingFor(new LogMessageWaitStrategy()
                            .withRegEx(".*database system is ready to accept connections.*\\s")
                            .withTimes(2)
                            .withStartupTimeout(Duration.of(60, SECONDS)))
                    .withCommand("postgres", "-c", "fsync=off")     //关闭文件同步
                    .withNetwork(network);
            this.workerContainers.add(citus);
        }
    }

    //使用静态方法创建容器对象,启动容器，执行容器中的脚本。默认放到资源目录下：src/main/resources
    public static Future<CitusCluster> create(Vertx vertx, String sqlPath) {
        FileSystem fs = vertx.fileSystem();

        //先将本地文件读入vert.x的文件系统，然后将vert.x的文件挂载到容器中的路径，接着启动容器，运行挂载到容器中的脚本文件
        return Future.<List<String>>future(promise -> fs.readDir(sqlPath, promise))
                .compose(hostScripts ->
                        Future.future(promise ->
                                vertx.executeBlocking(p -> {
                                    CitusCluster cluster = new CitusCluster();

                                    //挂载大量本地文件到容器中的某个目录下
                                    List<String> containerScripts = new ArrayList<>();
                                    hostScripts.forEach(hostScript -> {
                                        String containerScript = FilenameUtils.concat("/sql-import", FilenameUtils.getName(hostScript));
                                        containerScripts.add(containerScript);
                                        cluster.postgres.addFileSystemBind(hostScript, containerScript, BindMode.READ_ONLY);    //将两边的文件进行绑定
                                    });
                                    //将脚本文件按照字典顺序排序
                                    containerScripts.sort(String::compareTo);

                                    //启动容器
                                    cluster.start();

                                    //将容器中指定目录下的脚本执行--准备命令数组,然后执行
                                    ArrayList<String> cmdList = new ArrayList<String>();
                                    cmdList.add("psql");
                                    cmdList.add("-U");
                                    cmdList.add("postgres");
                                    containerScripts.forEach(containerScript -> {
                                        cmdList.add("-f");
                                        cmdList.add(containerScript);
                                    });

                                    System.out.println(cmdList);

                                    try {
                                        cluster.postgres.execInContainer(cmdList.toArray(new String[0]));
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }

                                    p.complete(cluster);
                                }, promise)
                        )
                );
    }


    @Override
    public void start() {
        //启动集群
        postgres.start();
        workerContainers.forEach(GenericContainer::start);

        //执行命令：master容器将复本容器进行关联
        workerContainers.forEach(container -> {
            try {
                String statement = String.format("SELECT * from master_add_node('%s', %d)", container.getNetworkAliases().get(0), 5432);
                postgres.execInContainer("psql", "-U", "postgres", "-c", statement);
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void stop() {
        //停止集群
        workerContainers.forEach(GenericContainer::stop);
        postgres.stop();
    }

    public GenericContainer getPostgres() {
        return postgres;
    }

    public void setPostgres(GenericContainer postgres) {
        this.postgres = postgres;
    }

    public List<GenericContainer> getWorkerContainers() {
        return workerContainers;
    }

    public void setWorkerContainers(List<GenericContainer> workerContainers) {
        this.workerContainers = workerContainers;
    }
}
